package com.example.spidertp.core;


import com.example.spidertp.core.mapper.TpWarningNoticeMapper;
import com.example.spidertp.core.model.entity.TpWarningNotice;
import com.example.spidertp.core.model.enums.TpWarningNoticeTypeEnum;
import com.example.spidertp.utils.DateUtils;
import com.example.spidertp.utils.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author xiong.bo
 * @version 1.0
 * @date 2022/9/13 9:21 下午
 */

@Slf4j
public class SpiderThreadPool extends ThreadPoolExecutor {

    //通知间隔时间
    private static final Long INTERVAL = 5 * 60 * 1000L;
    //任务超时时间（根据业务需要定义）
    private static final Long OVERTIME = 5 * 1000L;

    //阀值因子,目前固定阀值为80%
    private static final Float THRESHOLD_FACTOR = 0.8f;

    //通知的时间
    private Long maxPoolSizeWarningTime;
    private Long queueCapacityWarningTime;
    private Long taskExecOverTimeWarningTime;

    //线程任务执行时间
    private Long threadExecStartTime;

    //记录创建线程池时传入的工作队列的个数
    private int queueCapacity;

    private boolean isPause = false;

    private final ReentrantLock lock = new ReentrantLock();

    private final Condition condition = lock.newCondition();


    /**
     * @param corePoolSize    核心线程数
     * @param maximumPoolSize 最大线程数
     * @param keepAliveTime   空闲线程存活时间
     * @param unit            空闲线程存活时间单位
     * @param workQueue       工作队列
     */
    public SpiderThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    /**
     * @param corePoolSize  核心线程数
     * @param maxPoolSize   最大线程数
     * @param keepAliveTime 空闲线程存活时间
     * @param unit          空闲线程存活时间单位
     * @param workQueue     工作队列
     * @param handler       拒绝策略，当工作列队满了，并且线程数到达最大数量时触发的拒绝策略
     */
    public SpiderThreadPool(int corePoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    /**
     * @param corePoolSize  核心线程数
     * @param maxPoolSize   最大线程数
     * @param keepAliveTime 空闲线程存活时间
     * @param unit          空闲线程存活时间单位
     * @param workQueue     工作队列
     * @param threadFactory 线程池工厂
     * @param handler       拒绝策略，当工作列队满了，并且线程数到达最大数量时触发的拒绝策略
     */
    public SpiderThreadPool(int corePoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    public int getQueueCapacity() {
        return queueCapacity;
    }

    //仅是记录创建线程池时传入的工作队列的个数
    public void setQueueCapacity(int queueCapacity) {
        this.queueCapacity = queueCapacity;
    }


    /**
     * @return 获取线程池队排队的元素（动态变化,瞬时值）
     */
    public int getQueueTaskSize() {
        return this.getQueue().size();
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        lock.lock();
        try {
            logWarningNotice();
            threadExecStartTime = System.currentTimeMillis();

            super.beforeExecute(t, r);
            while (isPause){
                log.warn("thread pausing:{}\n", t.getName());
                condition.await();
            }
        } catch (Exception e) {
            log.error("SpiderThreadPool beforeExecute error!", e);
            Thread.currentThread().interrupt();
        } finally {
            lock.unlock();
        }
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        logWarningNotice();
    }

    /**
     * 暂停
     */
    public void pause() {
        lock.lock();
        try {
            isPause = true;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 恢复
     */
    public void resume() {
        lock.lock();
        try {
            isPause = false;
            condition.signalAll();
        } finally {
            lock.unlock();
        }
    }


    // 记录警告通知
    private void logWarningNotice() {

        int maximumPoolSize = this.getMaximumPoolSize();
        int currentPoolSize = this.getPoolSize();
        boolean maxPoolSizeWarningTimeFlag = maxPoolSizeWarningTime == null || (System.currentTimeMillis() - maxPoolSizeWarningTime) > INTERVAL;
        boolean queueCapacityWarningTimeFlag = queueCapacityWarningTime == null || (System.currentTimeMillis() - queueCapacityWarningTime) > INTERVAL;
        boolean taskExecOverTimeWarningTimeFlag = taskExecOverTimeWarningTime == null || (System.currentTimeMillis() - taskExecOverTimeWarningTime) > INTERVAL;
        Long threadExecEndTime = System.currentTimeMillis();

        TpWarningNotice capacityWn = null;
        TpWarningNotice maxSizeWn = null;
        TpWarningNotice overtimeWn = null;

        if ((float) currentPoolSize / maximumPoolSize > THRESHOLD_FACTOR && maxPoolSizeWarningTimeFlag) {
            maxSizeWn = new TpWarningNotice();
            maxSizeWn.setNoticeType(TpWarningNoticeTypeEnum.MAX_POOL_SIZE_WARNING.getCode());
            maxPoolSizeWarningTime = System.currentTimeMillis();
        }
        if ((float) this.getQueueTaskSize() / queueCapacity > THRESHOLD_FACTOR && queueCapacityWarningTimeFlag) {
            capacityWn = new TpWarningNotice();
            capacityWn.setNoticeType(TpWarningNoticeTypeEnum.QUEUE_CAPACITY_WARNING.getCode());
            queueCapacityWarningTime = System.currentTimeMillis();
        }
        if (threadExecStartTime != null && (threadExecEndTime - threadExecStartTime) > OVERTIME && taskExecOverTimeWarningTimeFlag) {
            overtimeWn = new TpWarningNotice();
            overtimeWn.setNoticeType(TpWarningNoticeTypeEnum.TASK_EXEC_OVERTIME_WARNING.getCode());
            taskExecOverTimeWarningTime = System.currentTimeMillis();
        }

        if (capacityWn != null) {
            saveWarningNotice(capacityWn);
        }
        if (maxSizeWn != null) {
            saveWarningNotice(maxSizeWn);
        }
        if (overtimeWn != null) {
            saveWarningNotice(overtimeWn);
        }
    }


    private void saveWarningNotice(TpWarningNotice wn) {
        String poolName = SpiderThreadPoolFactory.getPoolClassMap().get(String.valueOf(System.identityHashCode(this)));
        wn.setPoolName(poolName);
        wn.setCorePoolSize(this.getCorePoolSize());
        wn.setMaximumPoolSize(this.getMaximumPoolSize());
        wn.setCurrentPoolSize(this.getPoolSize());
        wn.setCurrentQueueSize(this.getQueueTaskSize());
        wn.setHisMaximumPoolSize(this.getLargestPoolSize());
        wn.setKeepAliveTime(this.getKeepAliveTime(TimeUnit.SECONDS));
        wn.setCreateUser("admin");
        wn.setCreateTime(DateUtils.getCurrentTimestamp());
        TpWarningNoticeMapper warningNoticeMapper = SpringContextUtil.getBean(TpWarningNoticeMapper.class);
        warningNoticeMapper.insert(wn);
    }

}
